package com.wzz.shop.realtime.etl.process

import com.alibaba.fastjson.JSON
import com.alibaba.fastjson.serializer.SerializerFeature
import com.wzz.shop.realtime.etl.`trait`.MysqlBaseETL
import com.wzz.shop.realtime.etl.bean.OrderDBEntity
import com.wzz.shop.realtime.etl.utils.GlobalConfigUtil
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}

/**
 *
 * @author ${author}
 * @date 2021-09-14 19:57
 *
 */
class OrderETL(env:StreamExecutionEnvironment) extends MysqlBaseETL(env){
  override def process(): Unit = {
    val kafkavalue = getKafkaDataStream()
    kafkavalue.printToErr("yuanshuju>>>>")
    //过滤找出添加类型的orders表
    val kafkavaluefilter = kafkavalue.filter(data => {
      data.getTableName.equals("wzz_orders") && data.getEventType.equals("insert")
    })
    //转换成bean对象
    val orderDbEntityValue = kafkavaluefilter.map(data => {
      OrderDBEntity(data)
    })
    //将bean对象转换成json类型
    val jsonorderDbEntityValue = orderDbEntityValue.map(data => {
      JSON.toJSONString(data, SerializerFeature.DisableCircularReferenceDetect)
    })
    jsonorderDbEntityValue.printToErr("订单解析数据》》")
    //写入到kafka的dwd_order的主题中
    jsonorderDbEntityValue.addSink(kafkaProducer(GlobalConfigUtil.`output.topic.order`))
  }


}
